package belf.migrate.api.sink.jdbc;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.sink.Sink;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.CatalogTable;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.FieldCase;
import belf.migrate.api.util.QuoteType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 * JDBC方式保存变动数据的基类，每个插件的{@link Sink}需继承自这个类
 */
@Slf4j
public abstract class JdbcSink implements Sink {
    private AbstractSinkCatalog catalog;

    private JobContext jobContext;

    private JdbcSinkWorker jdbcSinkWorker;

    public JdbcSink(JobContext jobContext) {
        this.jobContext = jobContext;
        this.catalog = jobContext.getSinkCatalog();
    }

    @Override
    public void setJobContext(JobContext jobContext) {
       this.jobContext = jobContext;
    }

    public AbstractSinkCatalog getCatalog() {
        return catalog;
    }

    public void setCatalog(AbstractSinkCatalog catalog) {
        this.catalog = catalog;
    }

    public JobContext getJobContext() {
        return jobContext;
    }

    @Override
    public List<CatalogTable> getCosumedCatalogTables() {
        return Sink.super.getCosumedCatalogTables();
    }

    /**
     * 将目标端所有表的信息都缓存起来，方便后面调用
     */
    @Override
    public void start() {
        try {
            // 打开目标端数据库连接
            catalog.open();

            List<String> tableNames = catalog.listTables(jobContext.getValidSinkDatabase(), jobContext.getValidSinkDatabase());
            String databaseName = jobContext.getTaskConf().getSinkConf().getDatabaseName();
            String schemaName = jobContext.getTaskConf().getSinkConf().getSchemaName();
            String catalogName = jobContext.getTaskConf().getSinkConf().getCatalogName();
            boolean isQuote = jobContext.getTaskConf().getSinkConf().isQuote();
            DatabaseType databaseType = DatabaseType.getDatabaseType(catalogName);
            String leftQuote = isQuote ? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
            String rightQuote = leftQuote;
            if (leftQuote.length() == 2) {
                rightQuote = leftQuote.substring(1);
                leftQuote = leftQuote.substring(0, 1);
            }
            FieldCase fieldCase = FieldCase.getFieldCase(jobContext.getTaskConf().getSinkConf().getFieldCase());
            QualifiersModel qualifiersModel = new QualifiersModel(leftQuote, rightQuote, fieldCase, databaseType.getQualifiedType());

            // 加载所有目标端表元数据，缓存在jobContext中
            for (String tableName : tableNames) {
                String simpleTableName = tableName;
                if (tableName.contains(databaseType.getQualifiedType().getValue())) {
                    simpleTableName = StringUtils.substringAfterLast(tableName, databaseType.getQualifiedType().getValue());
                }
                CatalogTable catalogTable = catalog.getTable(TablePath.of(databaseName, schemaName, simpleTableName, qualifiersModel));
                jobContext.addSinkTable(jobContext.getTaskConf().getSinkConf().getCatalogName(),
                        jobContext.getFullTableName(tableName),
                        catalogTable);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("Start JdbcSink error: {}", e.getMessage());
        }

        jdbcSinkWorker = new JdbcSinkWorker(jobContext);
        jdbcSinkWorker.start();
        log.info("Start JdbcSinkWorker-{}", jobContext.getJobId());
        //JVM退出时自动关闭数据库连接
        addShutdownHook();
    }

    private void addShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> stop()));
    }

    @Override
    public void stop() {
        //关闭目标端所有数据库连接
        try {
            jdbcSinkWorker.terminateThread();
            catalog.close();
        } catch (CatalogException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void pause() {
        jdbcSinkWorker.pause();
    }

    @Override
    public void resume() {
        jdbcSinkWorker.resumeThread();
    }
}
